use std::sync::{mpsc, Arc};
use std::thread;

use futures::{select, FutureExt};
use honeybee::prelude::*;
use honeybee::Result;

use crate::app_listen_delegate::AppListenDelegate;
use crate::config::CONFIG;
use crate::{room_manager, room_test};

pub fn run(shutdown_tx: mpsc::SyncSender<&'static str>) -> Addr<ShutdownActor> {
    let (tx, rx) = mpsc::sync_channel(1);
    let _ = thread::Builder::new()
        .name("async_main_thread".into())
        .spawn(move || {
            block_on(async move {
                let shutdown_addr = honeybee::prelude::init().await.unwrap();
                tx.send(shutdown_addr.clone()).unwrap();
                select! {
                    result = async_main().fuse() => {
                        if let Err(err) = result {
                            error!("async_main err:{}", err);
                        }
                    }
                    _ = shutdown_addr.wait_for_stop().fuse() => {}
                }
            });
            info!("async_main thread stoped");
            let _ = shutdown_tx.send("async_main exit");
        });
    rx.recv().unwrap()
}

async fn async_main() -> Result<()> {
    etcd::connect(&CONFIG.server.etcd_endpoints, None)
        .await
        .map_err(|err| {
            anyhow!(
                "connect etcd server:{:?} failed:{}",
                CONFIG.server.etcd_endpoints,
                err
            )
        })?;

    info!("try connect nats...");
    honeybee_nats::connect("nats://127.0.0.1:4222")
        .map_err(|err| anyhow!("connect nats server failed:{}", err))?;

    let _ = nats_subscribe::NatsSubscribeActor::new("room", proto::get_message_register())
        .start()
        .await?;

    let _ = TcpListenerActor::new(
        &CONFIG.node.node_type,
        &CONFIG.node.node_name,
        CONFIG.server.listen_port,
        proto::get_message_register(),
        Some(Arc::new(AppListenDelegate)),
    )
    .start()
    .await?;

    let etcd_usc_client = etcd_usc_client::EtcdUscClientActor::new(
        &CONFIG.node.node_type,
        &CONFIG.node.node_name,
        proto::get_message_register(),
    )
    .start()
    .await?;

    info!("wait usc server initialized...");
    etcd_usc_client.wait_for_stop().await;

    let _ = room_test::RoomTestActor::from_registry().await?;
    let _ = room_manager::RoomManager::from_registry().await?;

    futures::future::pending::<()>().await;
    Ok(())
}
